package com.jahe.learn.netty4.websocket;

import com.jahe.learn.netty4.ThreadPool;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * webscoket的客户端启动器。
 * ping包发送，接收pong包在Handler。
 * 断开自动重连。
 * @author: jahe.lai
 * @create: 2019-12-27 14:29
 */
public class WebSocketClient {

    private static Logger logger = LoggerFactory.getLogger(WebSocketClient.class);

    // 第一次启动成功是设置成 true. 用于做只需要初始化一次的任务做开关
    private static final AtomicBoolean START_SUCCESS = new AtomicBoolean(Boolean.FALSE);

    static final String WS_URI = String.format("ws://%s:%d/ws", "127.0.0.1", 8899);
    static URI uri;
    static final String CLIENT_SHAKE = "client_shake"; // 握手升级协议的处理
    private static Bootstrap bootstrap;
    static SocketChannel socketChannel;

    public static void main(String[] args) throws Exception {

        // 连接线程，监听连接状态的线程（重新连接），所以至少需要两个线程；
        // 只有一个线程，在重新连接时报异常：io.netty.util.concurrent.BlockingOperationException: DefaultChannelPromise@c0a4320(incomplete) at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:461)
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);

        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        //HTTP编解码器
                        pipeline.addLast("http_codec", new HttpClientCodec());
                        //HTTP消息聚合,使用FullHttpResponse和FullHttpRequest到ChannelPipeline中的下一个ChannelHandler，这就消除了断裂消息，保证了消息的完整。
                        pipeline.addLast("http_aggregator", new HttpObjectAggregator(65536));
                        pipeline.addLast("client_handler", new WSClientHandler());
                        pipeline.addLast(CLIENT_SHAKE, new WSClientShakeHandler());
                    }
                });
        uri = new URI(WS_URI);
        bootstrap.remoteAddress(uri.getHost(), uri.getPort());

        connection();
    }

    /**
     * 支持重新建立连接。
     */
    public static void connection() {
        try {
            ChannelFuture channelFuture = bootstrap.connect();

            // 针对服务器未启动时的重连。
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (!future.isSuccess()) {
                        reConnection(future.channel().eventLoop(), "ChannelFutureListener");
                    }
                }
            });

            // 等至连接成功。如果连接不成功，抛出异常，终止以下的流程。
            channelFuture.sync();

            socketChannel = (SocketChannel) channelFuture.channel();

            // 握手，协议升级
            handShake(uri, channelFuture);

            logger.warn("start success.");
            // 发送测试消息，带中文。
            socketChannel.writeAndFlush(new TextWebSocketFrame("来自client 的消息"));

            ping();
            START_SUCCESS.set(Boolean.TRUE);
            // 阻塞并监听关闭信号 （在收到 CloseWebSocketFrame 时，退出监听， 进而执行finally块代码。）
//            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            logger.warn("start-connection error", e);
        } finally {
            // 在连接断开时，要使用重新连接的话。这里不关闭线程池。
//            logger.warn("eventLoopGroup.shutdownGracefully()");
//            eventLoopGroup.shutdownGracefully();
        }
    }

    /**
     * 握手和协议升级
     * @param uri
     * @throws Exception
     */
    private static void handShake(URI uri, ChannelFuture future) throws Exception {
        if (!future.isSuccess()) {
            return;
        }
        HttpHeaders httpHeaders = new DefaultHttpHeaders();
        // 握手
        WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri,
                WebSocketVersion.V13, (String)null, true, httpHeaders);

        WSClientShakeHandler handler = (WSClientShakeHandler)socketChannel.pipeline().get(CLIENT_SHAKE);
        handler.setHandshaker(handshaker);

        // 通过它构造握手响应消息返回给客户端，
        // 同时将WebSocket相关的编码和解码类动态添加到ChannelPipeline中，用于WebSocket消息的编解码，
        // 添加WebSocketEncoder和WebSocketDecoder之后，服务端就可以自动对WebSocket消息进行编解码了
        handshaker.handshake(socketChannel);
        ChannelPromise channelPromise = handler.getHandshakeFuture().sync();
        logger.warn("handshaking success.");
    }

    /**
     * 重新连接
     * 1。解决启动时 服务器未启动 见客户端启动方法的调用
     * 2。解决断开时重连接 @see WsClientHandler#channelInActive(...)
     * @param loop
     */
    static void reConnection(final EventLoop loop, String invoker) {
        loop.schedule(new Runnable() {
            @Override
            public void run() {
                logger.warn(START_SUCCESS.get() ? invoker+"尝试重新建立连接......" : invoker+"尝试连接到服务器……");
                connection();
            }
        }, 5L, TimeUnit.SECONDS);
    }

    private static void ping() {
        // 只需要连接成功时，初始化一次任务。不要重复初始化心跳任务。
        if (START_SUCCESS.get()) {
            return;
        }
        final int initSec = 25, delaySec = 30;
        logger.info("init ping schedule TASK. {} 秒后开始Ping, 间隔 {} 秒。", initSec, delaySec);
        ThreadPool.schedule(new Runnable() {
            @Override
            public void run() {
                if (socketChannel.isActive()) {
//
                    ThreadPool.submit(new Runnable() {
//                        @Override
                        public void run() {
                            if (socketChannel.isActive()) {
                                ChannelFuture pingFuture = socketChannel.writeAndFlush(new PingWebSocketFrame(Unpooled.copiedBuffer("PING".getBytes())));
                                try {
                                    pingFuture.get(1, TimeUnit.SECONDS);
                                } catch (Exception e) {
                                    logger.warn("ping err", e);
                                }
                            }
                        }

                    });
                }

            }
        }, initSec, delaySec);
    }

    public static Channel getChannel(){
        return socketChannel;
    }
}
